package spike.s003.sqlflow;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import spike.s003.sqlflow.entity.*;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.groupingBy;
import static spike.s003.sqlflow.SQLFlowMakeService.EntityProducerBase.FINISH_READING_DB_DATA;

/**
 * Created by ylyang on 02/04/2017.
 */
@Service
public class SQLFlowMakeService {
    private static final int MAX_QUEUE_SIZE = 1000;
    private static final int MAX_NUM_OF_CONSUMER = 10;
    private static final int MAX_TIME_MS_TO_WAIT = 99999;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private MasterCourseService masterCourseService;

    public abstract class EntityProducerBase<EntityType> implements Runnable {
        public static final String FINISH_READING_DB_DATA = "FINISH_READING_DB_DATA";

        private ArrayBlockingQueue<QueueElem> queue = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);

        private boolean noElemFromNowOn = false;

        @Override
        public void run() {
            produce();
        }

        private void produce() {
            jdbcTemplate.query(obtainSelectSQL(), rs -> {
                putIntoQueue(rs);
            });

            putSomeFinishTAG();
        }

        protected void putIntoQueue(ResultSet rs) {
            try {
                EntityType entityItem = mapRowToEntity(rs);

                String uniqueCode = obtainUniqueCode(entityItem);

                QueueElem queueElem = new QueueElem();
                queueElem.entityItem = entityItem;
                queueElem.uniqueCode = uniqueCode;

                putIntoQueue(queueElem);
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        protected void putIntoQueue(QueueElem queueElem) {
            try {
                queue.put(queueElem);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        protected void putSomeFinishTAG() {
            QueueElem finishTAG = createFinishTAG();

            for (int i = 0; i < MAX_NUM_OF_CONSUMER; i++) {
                putIntoQueue(finishTAG);
            }
        }

        private QueueElem createFinishTAG() {
            QueueElem queueElem = new QueueElem();
            queueElem.uniqueCode = FINISH_READING_DB_DATA;
            return queueElem;
        }

        public abstract String obtainSelectSQL();

        public abstract EntityType mapRowToEntity(ResultSet rs) throws SQLException;

        public abstract String obtainUniqueCode(EntityType rs) throws SQLException;

        public boolean containsNoElemFromNowOn() {
            return noElemFromNowOn;
        }

        public class QueueElem {
            private String uniqueCode;
            private EntityType entityItem;

            public QueueElem() {

            }

            public String getUniqueCode() {
                return uniqueCode;
            }

            public EntityType getEntityItem() {
                return entityItem;
            }

            @Override
            public String toString() {
                return "QueueElem{" +
                        "uniqueCode='" + uniqueCode + '\'' +
                        ", entityItem=" + entityItem +
                        '}';
            }
        }

        public QueueElem pollQueue() {
            try {
                QueueElem queueElem = queue.poll(MAX_TIME_MS_TO_WAIT, TimeUnit.MILLISECONDS);
                if (null == queueElem) {
                    throw new RuntimeException();
                }

                String uniqueCode = queueElem.getUniqueCode();
                if (FINISH_READING_DB_DATA.equals(uniqueCode)) {
                    noElemFromNowOn = true;
                    return createFinishTAG();
                }

                return queueElem;
            } catch (InterruptedException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }

        public List<EntityType> collectEntities() {
            List<EntityType> entities = new ArrayList<>();

            while (true) {
                QueueElem queueElem = pollQueue();

                String uniqueCode = queueElem.getUniqueCode();
                if (FINISH_READING_DB_DATA.equals(uniqueCode)) {
                    break;
                }

                entities.add(queueElem.getEntityItem());
            }

            return entities;
        }
    }

    public class StudentProducer extends EntityProducerBase<StudentEntity> {
        @Override
        public String obtainSelectSQL() {
            return "select * from s003_students order by student_ID asc";
        }

        @Override
        public StudentEntity mapRowToEntity(ResultSet rs) throws SQLException {
            return StudentEntity.BeanRowMapper.mapRow(rs);
        }

        @Override
        public String obtainUniqueCode(StudentEntity studentEntity) throws SQLException {
            return studentEntity.getStudentID();
        }
    }

    public class StudentCourseProducer extends EntityProducerBase<StudentCourseEntity> {
        @Override
        public String obtainSelectSQL() {
            return "select * from s003_students_course_score order by student_ID asc";
        }

        @Override
        public StudentCourseEntity mapRowToEntity(ResultSet rs) throws SQLException {
            StudentCourseEntity studentCourseEntity = StudentCourseEntity.BeanRowMapper.mapRow(rs);

            String courseID = studentCourseEntity.getCourseID();
            CourseEntity courseEntity = masterCourseService.accessCourseByID(courseID);
            studentCourseEntity.setCourseEntity(courseEntity);

            return studentCourseEntity;
        }

        @Override
        public String obtainUniqueCode(StudentCourseEntity studentCourseEntity) throws SQLException {
            return studentCourseEntity.getStudentID();
        }
    }

    public class StudentQQProducer extends EntityProducerBase<StudentQQEntity> {
        @Override
        public String obtainSelectSQL() {
            return "select * from s003_students_qqs order by student_ID asc";
        }

        @Override
        public StudentQQEntity mapRowToEntity(ResultSet rs) throws SQLException {
            return StudentQQEntity.BeanRowMapper.mapRow(rs);
        }

        @Override
        public String obtainUniqueCode(StudentQQEntity studentQQEntity) throws SQLException {
            return studentQQEntity.getStudentID();
        }
    }


    public StudentProducer createStudentProducer() {
        return new StudentProducer();
    }

    public StudentCourseProducer createStudentCourseProducer() {
        return new StudentCourseProducer();
    }

    public StudentQQProducer createStudentQQProducer() {
        return new StudentQQProducer();
    }

    public class StudentJsonProducer extends StudentProducer {
        @Override
        public void run() {
            StudentProducer studentProducer = createStudentProducer();
            StudentCourseProducer studentCourseProducer = createStudentCourseProducer();
            StudentQQProducer studentQQProducer = createStudentQQProducer();

            Arrays.<Runnable>asList(studentProducer, studentCourseProducer, studentQQProducer).forEach(
                    runnable -> new Thread(runnable).start()
            );

            List<QueueElem> studentElemBatch = new ArrayList<>();

            ProducerTuple<StudentCourseEntity> courseTuple = new ProducerTuple<StudentCourseEntity>(studentCourseProducer, studentCourseProducer.pollQueue());
            ProducerTuple<StudentQQEntity> qqTuple = new ProducerTuple<StudentQQEntity>(studentQQProducer, studentQQProducer.pollQueue());
            while (!studentProducer.containsNoElemFromNowOn()) {
                QueueElem studentElem = studentProducer.pollQueue();

                String uniqueCode = studentElem.getUniqueCode();
                if (FINISH_READING_DB_DATA.equals(uniqueCode)) {
                    putAllIntoQueue(studentElemBatch);
                    putSomeFinishTAG();
                    break;
                }

                StudentEntity studentEntity = studentElem.getEntityItem();

                if (courseTuple.containsElem()) {
                    List<StudentCourseEntity> studentCourseEntities = calcMembersOnStudentEntity(studentEntity, courseTuple);
                    if (!CollectionUtils.isEmpty(studentCourseEntities)) {
                        studentEntity.setStudentCourseEntities(studentCourseEntities);
                    }
                }

                if (qqTuple.containsElem()) {
                    List<StudentQQEntity> studentQQEntities = calcMembersOnStudentEntity(studentEntity, qqTuple);
                    if (!CollectionUtils.isEmpty(studentQQEntities)) {
                        studentEntity.setStudentQQEntities(studentQQEntities);
                    }
                }

                studentElemBatch.add(studentElem);

                if (studentElemBatch.size() >= 1000) {
                    putAllIntoQueue(studentElemBatch);

                    studentElemBatch = new ArrayList<>();
                }
            }
        }

        private class ProducerTuple<EntityType> {
            private EntityProducerBase<EntityType> producer;
            private EntityProducerBase<EntityType>.QueueElem peekOfQueue;

            private ProducerTuple(EntityProducerBase<EntityType> producer, EntityProducerBase<EntityType>.QueueElem peekOfQueue) {
                this.producer = producer;
                this.peekOfQueue = peekOfQueue;
            }

            public EntityProducerBase<EntityType>.QueueElem getPeekOfQueue() {
                return peekOfQueue;
            }

            public void setPeekOfQueue(EntityProducerBase<EntityType>.QueueElem peekOfQueue) {
                this.peekOfQueue = peekOfQueue;
            }

            public EntityProducerBase<EntityType> getProducer() {
                return producer;
            }

            public boolean containsElem() {
                return !producer.containsNoElemFromNowOn() || !FINISH_READING_DB_DATA.equals(peekOfQueue.getUniqueCode());
            }
        }

        private <EntityType> List<EntityType> calcMembersOnStudentEntity(StudentEntity studentEntity, ProducerTuple<EntityType> producerTuple) {
            List<EntityType> entities = new ArrayList<>();

            do {
                if (FINISH_READING_DB_DATA.equals(producerTuple.getPeekOfQueue().getUniqueCode())) {
                    break;
                } else {
                    int n = studentEntity.getStudentID().compareTo(producerTuple.getPeekOfQueue().getUniqueCode());
                    if (n == 0) {
                        entities.add(producerTuple.getPeekOfQueue().getEntityItem());
                    } else if (n < 0) {
                        break;
                    }
                }

                producerTuple.setPeekOfQueue(producerTuple.getProducer().pollQueue());
            } while (!producerTuple.getProducer().containsNoElemFromNowOn());

            return entities;
        }

        private void putAllIntoQueue(List<QueueElem> studentElemBatch) {
            putQQEntityOnStudents(studentElemBatch);

            for (QueueElem studentElem : studentElemBatch) {
                putIntoQueue(studentElem);
            }
        }

        private void putQQEntityOnStudents(List<QueueElem> studentElemBatch) {
            List<StudentQQEntity> studentQQEntities = new ArrayList<>();
            studentElemBatch.stream().map(e -> e.getEntityItem().getStudentQQEntities())
                    .filter(entities -> !CollectionUtils.isEmpty(entities))
                    .forEach(entities -> studentQQEntities.addAll(entities));

            if (!CollectionUtils.isEmpty(studentQQEntities)) {
                String sqlLine = "select * from s003_qqs" + " where qq_num in "
                        + "('"
                        + String.join("','", studentQQEntities.stream().map(q -> q.getQqNum()).collect(Collectors.toList()))
                        + "')";

                List<QQEntity> qqEntities = jdbcTemplate.query(sqlLine, new QQEntity.BeanRowMapper());

                if (!CollectionUtils.isEmpty(qqEntities)) {
                    Map<String, QQEntity> qqEntitiesMap = qqEntities
                            .stream().collect(groupingBy(QQEntity::getQqNum))
                            .entrySet()
                            .stream()
                            .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue().get(0)));

                    studentQQEntities.forEach(studentQQEntity -> {
                        studentQQEntity.setQqEntity(qqEntitiesMap.get(studentQQEntity.getQqNum()));
                    });
                }
            }
        }
    }

    public StudentJsonProducer createStudentJsonProducer() {
        return new StudentJsonProducer();
    }
}
